#include "msg_queue.h"

int init_queue(struct msg_queue *q, char *name, int maxsize){
	q->msg = NULL;
	set_queue_name(q, name);
	q->size = maxsize;
	q->lget = 0;
	q->lput = 0;
	q->nData = 0;
	q->nFullThread = 0;
	q->nEmptyThread = 0;
	pthread_mutex_init(&(q->mux), 0);
    pthread_cond_init(&(q->cond_get), 0);
    pthread_cond_init(&(q->cond_put), 0);
	
	q->msg = malloc(maxsize * sizeof(struct msg_buffer*));
	printf("msg_buffer size:%ld\n", sizeof(struct msg_buffer*));
    if(q->msg == NULL){
        printf("malloc failed!\n");
        return (-1);
    }
	return 0;
}

void destroy_queue(struct msg_queue *q){
	free(q->msg);
    pthread_mutex_destroy(&(q->mux));
    pthread_cond_destroy(&(q->cond_get));
    pthread_cond_destroy(&(q->cond_put));
}

struct msg_buffer* get_queue(struct msg_queue *q){
    struct msg_buffer* data = NULL;
    pthread_mutex_lock(&(q->mux));
    while(q->lget == q->lput && 0 == q->nData){
        // 此处循环判断的原因是：假设2个消费者线程在get_queue阻塞，然后两者都被激活，
        // 而其中一个线程运行比较块，快速消耗了2个数据，另一个线程醒来的时候已
        // 经没有新数据可以消耗了。这种情况是有可能的：比如，其它生产者线程快速
        // 调用put_queue两次，如果有2个线程在get_queue处阻塞，就会被同时激活，
        // 而完全有可能，其中一个被激活的线程获取到了cpu，快速处理了2个消息。

        // 对于循环队列，如果lget与lput相等，那么只有两种情况，
        // 1：nData不为0，队列满
        // 2：nData为0，队列空
        q->nEmptyThread++;
        pthread_cond_wait(&(q->cond_get), &(q->mux));
        q->nEmptyThread--;
    }
#ifdef DEBUG
    printf("get data! lget:%d", q->lget);
#endif
    data = (q->msg)[q->lget++];
    if(q->lget == q->size){
        // queue用作循环队列
        q->lget = 0;
    }
    q->nData--;
#ifdef DEBUG
    printf(" nData:%d\n", q->nData);
#endif
    if(q->nFullThread){
        // 仅在必要时才调用pthread_cond_signal, 尽量少陷入内核态
        pthread_cond_signal(&(q->cond_get));
    }
    pthread_mutex_unlock(&(q->mux));
    return data;
}

void put_queue(struct msg_queue *q, struct msg_buffer* data){
    pthread_mutex_lock(&(q->mux));
    while(q->lget == q->lput && q->nData){
        q->nFullThread++;
        pthread_cond_wait(&(q->cond_get), &(q->mux));
        q->nFullThread--;
    }
#ifdef DEBUG
    printf("put data! lput:%d", q->lput);
#endif
    (q->msg)[q->lput++] = data;
    if(q->lput == q->size){
        q->lput = 0;
    }
    q->nData++;
#ifdef DEBUG
    printf(" nData:%d\n", q->nData);
#endif
    if(q->nEmptyThread){
        pthread_cond_signal(&(q->cond_get));
    }
    pthread_mutex_unlock(&(q->mux));
}

void set_queue_name(struct msg_queue *q, char *name){
	strncpy(q->name, name, 255);
	q->name[255] = 0;
}

char *get_queue_name(struct msg_queue *q){
	return q->name;
}
